package com.sprucetec.live.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.remoting.exception.RemotingException;
import com.google.common.collect.Lists;
import com.sprucetec.live.common.dto.PageDTO;
import com.sprucetec.live.common.dto.PagedListDTO;
import com.sprucetec.live.common.dto.RedisCacheDto;
import com.sprucetec.live.common.dto.business.LiveChatDto;
import com.sprucetec.live.common.dto.business.LiveUserDto;
import com.sprucetec.live.common.dto.business.LiveVideoDto;
import com.sprucetec.live.constants.MqConstant;
import com.sprucetec.live.constants.RedisKeyConstant;
import com.sprucetec.live.entity.LiveSystemLog;
import com.sprucetec.live.entity.LiveUser;
import com.sprucetec.live.entity.LiveVideo;
import com.sprucetec.live.enums.ChatTypeEnum;
import com.sprucetec.live.service.RedisService;
import com.sprucetec.live.utils.RedisUtil;
import com.sprucetec.mcq.producer.impl.MessageProducer;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.connection.DataType;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @ClassName RedisServiceImpl
 * @Description:
 * @Author Kenny
 * @Date 2020/3/26
 **/
@Slf4j
@Service("redisService")
public class RedisServiceImpl implements RedisService {
    private static final String redisKeyPre = "LIVE:";

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;


    @Autowired
    private MessageProducer mcqMessageProduct;

    @Override
    public void set(String key, Object value, TimeUnit timeUnit, long timeout) {
        redisTemplate.opsForValue().set(key, value, timeout, timeUnit);
    }

    @Override
    public Object get(String key) {
        return redisTemplate.opsForValue().get(key);
    }


    /***************************************** New Begin***********************************/


    @Autowired
    private RedisUtil redisUtil;

    @Qualifier("supportThreadPool")
    @Autowired
    private ThreadPoolTaskExecutor supportThreadPool;

    /**
     * 直播viodeDto数据
     * 点击详情，更新统计视频数据
     *
     * @param liveVideoDto
     */
    @Override
    public void videoDetailData(LiveVideoDto liveVideoDto) {
        supportThreadPool.execute(() -> {
            String liveNo = liveVideoDto.getLiveNo();
            String uid = liveVideoDto.getUid();
            String onlineUserKey = RedisKeyConstant.LiveVideoDataStatistics.onlineUser + liveNo;
            String topNumKey = RedisKeyConstant.LiveVideoDataStatistics.topNum + liveNo;
            String watchNumKey = RedisKeyConstant.LiveVideoDataStatistics.watchNum + liveNo;
            //Object o = redisTemplate.boundHashOps(onlineUserKey).get(RedisKeyConstant.LiveVideoDataStatistics.onlineUser + uid);

            // 保存在线人数
            LiveUserDto userDto = new LiveUserDto();
            userDto.setUserCode(liveVideoDto.getUid());
            userDto.setUserName(liveVideoDto.getUsername());
            userDto.setCreateTime(new Date());
            userDto.setUserLoginIp(liveVideoDto.getIp());
            userDto.setLiveNo(liveVideoDto.getLiveNo());
            redisTemplate.boundHashOps(onlineUserKey).putIfAbsent(RedisKeyConstant.LiveVideoDataStatistics.onlineUser + uid, userDto);
            // 处理峰值人数
            long online = redisTemplate.boundHashOps(onlineUserKey).size();
            Object top = redisTemplate.boundValueOps(topNumKey).get(0, -1);
            if (top != null && StringUtils.isNotBlank(top.toString()) && (online > Long.parseLong(top.toString()))) {
                // 峰值人数
                redisTemplate.boundValueOps(topNumKey).increment(1);
            } else if (top != null && "".equals(top.toString())) {
                redisTemplate.boundValueOps(topNumKey).increment(1);
            }
            // 处理观看次数
            redisTemplate.boundValueOps(watchNumKey)
                    .increment(1);
        });
    }


    @Override
    public void videoDetailAway(LiveVideoDto liveVideoDto) {
        redisTemplate.boundHashOps(RedisKeyConstant.LiveVideoDataStatistics.onlineUser + liveVideoDto.getLiveNo())
                .delete(RedisKeyConstant.LiveVideoDataStatistics.onlineUser + liveVideoDto.getUid());
        // 通知用户下线
        //  todo 下面的需求不确定要不要做，先不写了， 该功能是通知各个客户端，当前某某某离开了
        /*Message message = new Message(MqConstant.TOPIC, MqConstant.VideoDataMq.USER_AWAY,
                MqConstant.VideoDataMq.KEY + new Date().getTime(), JSON.toJSONString(liveVideoDto).getBytes(StandardCharsets.UTF_8));
        try {
            mcqMessageProduct.send(message);
        } catch (Exception e) {
            log.error("mcq通知用户下线失败=>", e);
        }*/
    }


    /**
     * 视频数据变化更新相关视频信息
     *
     * @param liveVideo
     */
    @Override
    public void videoChangeRefreshCache(LiveVideo liveVideo) {
        Collection emptyRedisKey = Lists.newArrayList();
        emptyRedisKey.add(RedisKeyConstant.LiveKind.TOPVIDEO);
        emptyRedisKey.add(RedisKeyConstant.LiveKind.NEWLAST);
        emptyRedisKey.add(RedisKeyConstant.LiveKind.CAROUSEL);
        redisTemplate.delete(emptyRedisKey);
        // 删除单个文章的redis缓存
        log.info("设置文章，并更新缓存{}" + liveVideo.getLiveNo());
        redisTemplate.boundHashOps(RedisKeyConstant.LiveVideoResources.VIDEO_DETAIL).delete(liveVideo.getLiveNo());
    }

    /**
     * 统计聊天人数 聊天次数 提问人数 提问次数
     *
     * @param liveChatDto
     */
    @Override
    public void chatDataOnVideo(LiveChatDto liveChatDto) {

        if (ChatTypeEnum.CHAT.getCode().equals(liveChatDto.getChatType())) {
            // 聊天人数
            redisTemplate.boundSetOps(RedisKeyConstant.LiveVideoDataStatistics.chatPersonNum + liveChatDto.getLiveNo())
                    .add(liveChatDto.getCreateUserCode());
            // 聊天次数
            redisTemplate.boundValueOps(RedisKeyConstant.LiveVideoDataStatistics.chatNum + liveChatDto.getLiveNo())
                    .increment(1);
        } else {
            // 提问人数
            redisTemplate.boundSetOps(RedisKeyConstant.LiveVideoDataStatistics.questionPersonNum + liveChatDto.getLiveNo())
                    .add(liveChatDto.getCreateUserCode());

            // 提问次数
            redisTemplate.boundValueOps(RedisKeyConstant.LiveVideoDataStatistics.questionNum + liveChatDto.getLiveNo())
                    .increment(1);
        }
    }

    @Override
    public PagedListDTO<RedisCacheDto> getRedisCachePage(RedisCacheDto redisCacheDto) {
        PagedListDTO pagedListDTO = new PagedListDTO();
        int pageIndex = redisCacheDto.getPageIndex();
        int pageSize = redisCacheDto.getPageSize();
        pagedListDTO.setPage(new PageDTO(pageIndex, pageSize));
        Integer from = (pageIndex - 1) * pageSize;
        if (StringUtils.isNotBlank(redisCacheDto.getKey())) {
            List<RedisCacheDto> list = new ArrayList<>();


            Set<String> keys = redisTemplate.keys(redisCacheDto.getKey());
            if (keys.size() != 0) {
                RedisCacheDto rd = new RedisCacheDto();
                rd.setKey(redisCacheDto.getKey());
                list.add(rd);
            }
            pagedListDTO.setTotal(1);
            pagedListDTO.setRecords(list);

            return pagedListDTO;
        }
        Set<String> lives = redisTemplate.keys(redisKeyPre + "*");
        List<RedisCacheDto> collect = lives.stream().map((key) -> {
            RedisCacheDto rd = new RedisCacheDto();
            DataType type = redisTemplate.type(key);

            if (type.code().equals(DataType.STRING.code())) {
                rd.setValue(JSONObject.toJSONString(redisTemplate.boundValueOps(key).get()));
            } else if (type.code().equals(DataType.HASH.code())) {
                rd.setValue(JSONArray.toJSONString(redisTemplate.boundHashOps(key).values()));
            } else if (type.code().equals(DataType.LIST.code())) {
                rd.setValue(JSONArray.toJSONString(redisTemplate.boundListOps(key).range(0, -1)));
            }
            rd.setKey(key);
            return rd;
        }).collect(Collectors.toList());
        pagedListDTO.setTotal(collect.size());
        try {
            pagedListDTO.setRecords(collect.subList(from, from + pageSize));
        } catch (Exception ex) {
            pagedListDTO.setRecords(collect.subList(from, collect.size()));
        }
        return pagedListDTO;
    }

    @Override
    public Integer deleteCache(RedisCacheDto redisCacheDto) {
        redisTemplate.delete(redisCacheDto.getKey());
        return 1;
    }

    /**
     * 清空所有redis缓存
     *
     * @param redisCacheDto
     * @return
     */
    @Override
    public Integer clearRedis(RedisCacheDto redisCacheDto) {
        Set<String> lives = redisTemplate.keys(redisKeyPre + "*");
        log.info("清空所有缓存");
        redisTemplate.delete(lives);
        return 1;
    }


}
